package org.nbict.iot.task.realtime.state;

import com.google.common.collect.Lists;
import org.apache.storm.mongodb.common.QueryFilterCreator;
import org.apache.storm.trident.operation.TridentCollector;
import org.apache.storm.trident.state.State;
import org.apache.storm.trident.tuple.TridentTuple;
import org.bson.Document;
import org.bson.conversions.Bson;
import org.nbict.iot.trident.mogo.MongoDBClient;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

/**
 * 状态替换
 * Created by songseven on 18/6/29.
 */
public class Timely implements State {

    private MongoDBClient mongoClient;
    private Options options;
    private Comparer transfer;
    private QueryFilterCreator query;


    public Timely() {
    }

    public Timely(Options options, Comparer transfer, QueryFilterCreator query) {
        this.options = options;
        this.transfer = transfer;
        this.query = query;
    }

    protected void prepare() {
        this.mongoClient = new MongoDBClient(this.options.url, this.options.database, this.options.username, this.options.password, this
                .options.admin, this.options.collectionName);
    }

    @Override
    public void beginCommit(Long txid) {

    }

    @Override
    public void commit(Long txid) {

    }

//    public Timely id(String id) {
//        this.id = id;
//        return this;
//    }

    /**
     * 更新各维度
     * @param tuples
     * @param collector
     */
    public void replaceTimelyDimension(List<TridentTuple> tuples, TridentCollector collector) {
        ArrayList documents = Lists.newArrayList();
        Iterator iterator = tuples.iterator();

        while (iterator.hasNext()) {
            TridentTuple tuple = null;
            tuple = (TridentTuple)iterator.next();
            Bson query = this.query.createFilter(tuple);
            Document replaced = this.mongoClient.get(query);
            Document res = this.transfer.compareAndSet(tuple, replaced);
            if (res != null) {
                mongoClient.replace(query, res, true);
            }
        }
    }

    public static class Options implements Serializable {
        private String url;
        private String database;
        private String username;
        private String password;
        private String admin;
        private String collectionName;
//        private Comparer transfer;

        public Options() {
        }

        public Timely.Options withUrl(String url) {
            this.url = url;
            return this;
        }

        public Timely.Options withDatabase(String database) {
            this.database = database;
            return this;
        }

        public Timely.Options withUsername(String username) {
            this.username = username;
            return this;
        }

        public Timely.Options withPassword(String password) {
            this.password = password;
            return this;
        }

        public Timely.Options withAdmin(String admin) {
            this.admin = admin;
            return this;
        }

        public Timely.Options withCollectionName(String collectionName) {
            this.collectionName = collectionName;
            return this;
        }
//
//        public Timely.Options withTransfer(Comparer transfer) {
//            this.transfer = transfer;
//            return this;
//        }
    }
}
